﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading.Actors.Languages;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;


namespace System.Threading.Actors
{
    // Handles isolated "sub-state" for each actor in a multi-tenant environment
    // Typically used in conjunction with StateBufferingHelper to manage multiple isolated actor states.
    public class IsolatedActorState : IActorState
    {
        public string PartitionName { get; private set; }
        internal readonly StateBufferingHelper m_aggregatedActorState;
        private readonly MethodCallHelper m_methodCallHelper;
        private readonly PubSubHelper m_pubSubHelper;
        private readonly IActorInteractor m_actorInteractionHelper;
        private readonly AssemblyHelper m_assemblyHelper;
        private readonly string m_serviceName;
        private readonly ConcurrentDictionary<string, object> m_permaStore;
        private readonly SemaphoreSlim m_asyncStateLock = new SemaphoreSlim(1, 1);
        private readonly ActorETWEventListener m_ETWListener;

        public IsolatedActorState(StateBufferingHelper aggregatedActorState, string partitionName, string serviceName, ActorETWEventListener eventListener, IActorInteractor actorInteractor)
        {
            Contract.Requires(aggregatedActorState != null);
            Contract.Requires(partitionName != null);
            // Contract.Requires(actorInteractor != null); // Unit tests violate this due to layering concerns.

            PartitionName = partitionName;
            m_aggregatedActorState = aggregatedActorState; // coordinator for multiple isolated states.
            m_permaStore = aggregatedActorState.GetPermaStore(partitionName); // the durable in-memory store for this isolated state
            m_methodCallHelper = new MethodCallHelper(this); // helper logic for making method calls
            m_ETWListener = eventListener; // Logging
            m_pubSubHelper = new PubSubHelper(this, partitionName, eventListener); // helper logic for pub/sub ops 
            m_actorInteractionHelper = actorInteractor; // helper logic for creating/deleting other actors
            m_assemblyHelper = new AssemblyHelper(this, m_permaStore, m_methodCallHelper); // helper logic for assembly-handling
            m_assemblyHelper.AddAssemblyResolver();
            m_serviceName = serviceName;
            // Register ourself with the stateMapper
            ActorProxyBase.RegisterState(serviceName, partitionName, this);

            // For now, just plug Javascript/Python support into every IsolatedActorState
            AddLanguagePack(SupportedLanguages.Javascript, () => new JavascriptLanguagePackage(this));
            AddLanguagePack(SupportedLanguages.Python, () => new PythonLanguagePackage(this));
        }

        public SemaphoreSlim AsyncStateLock { get { return m_asyncStateLock; } }

        internal void SetDurableValue(string key, object value, string source, bool scour)
        {
            // Check to see whether or not this item is an assembly, and act accordingly
            if (!m_assemblyHelper.CheckForAssemblyOrMethod(key, value, scour))
            {
                m_permaStore[key] = value;
            }
        }

        internal void RemoveDurableValue(string key)
        {
            object temp;
            m_permaStore.TryRemove(key, out temp);
        }

        public void Set(string key, object value)
        {
            m_aggregatedActorState.Set(PartitionName, key, value);
        }

        public void Remove(string key)
        {
            m_aggregatedActorState.Remove(PartitionName, key);
        }

        public object Get(string key)
        {
            return m_aggregatedActorState.Get(PartitionName, key);
        }

        public bool TryGet(string key, out object value)
        {
            return m_aggregatedActorState.TryGet(PartitionName, key, out value);
        }


        public Task Flush()
        {
            return m_aggregatedActorState.Flush(PartitionName);
        }

        public bool ChangesPendingForCurrentOperation()
        {
            return m_aggregatedActorState.ChangesPendingForCurrentOperation(PartitionName);
        }

        public void ClearChangesForCurrentOperation()
        {
            m_aggregatedActorState.ClearChangesForCurrentOperation(PartitionName);
        }

        public IEnumerable<string> GetAllKeysWithPrefix(string keyPrefix)
        {
            return m_aggregatedActorState.GetAllKeysWithPrefix(PartitionName, keyPrefix);
        }

        public object CallMethod(string methodName, object[] args)
        {
            return m_methodCallHelper.CallMethod(methodName, args);
        }

        public void Subscribe(IPublicationCallbackContract callerProxy, string eventType)
        {
            m_pubSubHelper.Subscribe(callerProxy, eventType);
        }

        public void Unsubscribe(IPublicationCallbackContract callerProxy, string eventType)
        {
            m_pubSubHelper.Unsubscribe(callerProxy, eventType);
        }

        public void UnsubscribeAll(IPublicationCallbackContract callerProxy)
        {
            if (callerProxy == null)
                m_pubSubHelper.UnsubscribeEverything();
            else
                m_pubSubHelper.UnsubscribeAll(callerProxy);
        }

        public void Publish(string eventType, object eventPayload)
        {
            m_pubSubHelper.Publish(eventType, eventPayload);
        }

        public void Publish(string eventType, object eventPayload, params object[] eventPayloadQualifiers)
        {
            m_pubSubHelper.Publish(eventType, eventPayload, eventPayloadQualifiers);
        }

        public IActorProxy GetActorProxy(string externalActorName, string partitionName = null)
        {
            Contract.Assert(m_actorInteractionHelper != null, "You must pass in an IActorInteractor to IsolatedActorState's constructor");
            return m_actorInteractionHelper.GetActorProxy(externalActorName, this, partitionName, PartitionName);
        }

        public IActorPromise GetActorProxyAsync(string externalActorName, string partitionName = null)
        {
            var proxyTask = m_actorInteractionHelper.GetActorProxyAsync(externalActorName, this, partitionName, PartitionName);
            return new ActorPromise(this, proxyTask);
        }

        //public IActorPromise Request(string actorName, string partitionKeyName, string methodName, params object[] parameters)
        //{
        //    var computationTask = Task.Run(async () =>
        //        {
        //            var proxy = await m_actorInteractionHelper.GetActorProxyAsync(actorName, this, partitionKeyName, this.PartitionName);
        //            proxy.Request(methodName, parameters);
        //        });
        //    var proxyTask = m_actorInteractionHelper.GetActorProxyAsync(actorName, this, partitionKeyName, this.PartitionName);
        //    var methodCallTask = proxyTask.ContinueWith(antecedent =>
        //    {
        //        return antecedent.Result.Request(methodName, parameters);
        //    }, TaskContinuationOptions.ExecuteSynchronously);
        //    return methodCallTask.Un
        //}

        public string GetId()
        {
            return m_serviceName;
        }

        public void AddLanguagePack(SupportedLanguages language, Func<ILanguagePackage> pack)
        {
            m_methodCallHelper.AddLanguagePack(language, pack);
        }

        public void AddMethod(SupportedLanguages language, string methodName, string methodImpl)
        {
            m_methodCallHelper.AddMethod(language, methodName, methodImpl);
        }

        public bool CreateActor(string actorName, string creationString, int numPartitions = 1)
        {
            Contract.Assert(m_actorInteractionHelper != null, "You must pass in an IActorInteractor to IsolatedActorState's constructor");
            return m_actorInteractionHelper.CreateActor(actorName, creationString, numPartitions);
        }

        public void DeleteActor(string actorName)
        {
            Contract.Assert(m_actorInteractionHelper != null, "You must pass in an IActorInteractor to IsolatedActorState's constructor");
            m_actorInteractionHelper.DeleteActor(actorName);
        }

        public void Dispose()
        {
            if (m_assemblyHelper != null) m_assemblyHelper.RemoveAssemblyResolver();
            if (m_pubSubHelper != null) m_pubSubHelper.Close();
            if (m_actorInteractionHelper != null) m_actorInteractionHelper.Dispose();
            if (m_serviceName != null) ActorProxyBase.UnregisterState(m_serviceName, PartitionName);
            // We are sharing the ActorETWEventListener instance.  Don't close it.
        }

        public void Configure(string configurationString)
        {
            ConfigurationHelper.Configure(configurationString, m_ETWListener);
        }
    }
}
